"""
事件总线 - 统一的事件发布订阅机制

这个模块实现了事件驱动架构的核心组件，支持：
- 事件发布和订阅
- 事件持久化
- 事件路由和过滤
- 异步事件处理
"""

import asyncio
import json
import logging
import threading
import time
import uuid
from abc import ABC, abstractmethod
from dataclasses import dataclass, asdict
from enum import Enum
from typing import Any, Callable, Dict, List, Optional, Set, Union
from datetime import datetime, timezone

logger = logging.getLogger(__name__)


class EventType(Enum):
    """事件类型枚举"""
    TASK_CREATED = "task_created"
    TASK_ASSIGNED = "task_assigned"
    TASK_STARTED = "task_started"
    TASK_COMPLETED = "task_completed"
    TASK_FAILED = "task_failed"
    TASK_CANCELLED = "task_cancelled"
    TASK_TIMEOUT = "task_timeout"
    
    SERVER_ONLINE = "server_online"
    SERVER_OFFLINE = "server_offline"
    SERVER_OVERLOADED = "server_overloaded"
    SERVER_UNDERLOADED = "server_underloaded"
    
    SCALE_OUT_REQUESTED = "scale_out_requested"
    SCALE_IN_REQUESTED = "scale_in_requested"
    SCALE_OUT_COMPLETED = "scale_out_completed"
    SCALE_IN_COMPLETED = "scale_in_completed"
    
    WEBHOOK_SENT = "webhook_sent"
    WEBHOOK_FAILED = "webhook_failed"
    WEBHOOK_RETRY = "webhook_retry"
    
    SYSTEM_ERROR = "system_error"
    SYSTEM_WARNING = "system_warning"
    SYSTEM_INFO = "system_info"


@dataclass
class Event:
    """事件数据结构"""
    id: str
    type: EventType
    source: str
    timestamp: datetime
    data: Dict[str, Any]
    correlation_id: Optional[str] = None
    metadata: Optional[Dict[str, Any]] = None
    
    def to_dict(self) -> Dict[str, Any]:
        """转换为字典格式"""
        return {
            "id": self.id,
            "type": self.type.value,
            "source": self.source,
            "timestamp": self.timestamp.isoformat(),
            "data": self.data,
            "correlation_id": self.correlation_id,
            "metadata": self.metadata or {}
        }
    
    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'Event':
        """从字典创建事件"""
        return cls(
            id=data["id"],
            type=EventType(data["type"]),
            source=data["source"],
            timestamp=datetime.fromisoformat(data["timestamp"]),
            data=data["data"],
            correlation_id=data.get("correlation_id"),
            metadata=data.get("metadata", {})
        )


class EventHandler(ABC):
    """事件处理器抽象基类"""
    
    @abstractmethod
    async def handle(self, event: Event) -> None:
        """处理事件"""
        pass
    
    @abstractmethod
    def can_handle(self, event_type: EventType) -> bool:
        """判断是否能处理指定类型的事件"""
        pass


class EventFilter:
    """事件过滤器"""
    
    def __init__(self, 
                 event_types: Optional[Set[EventType]] = None,
                 sources: Optional[Set[str]] = None,
                 correlation_ids: Optional[Set[str]] = None,
                 custom_filter: Optional[Callable[[Event], bool]] = None):
        self.event_types = event_types
        self.sources = sources
        self.correlation_ids = correlation_ids
        self.custom_filter = custom_filter
    
    def matches(self, event: Event) -> bool:
        """检查事件是否匹配过滤器"""
        if self.event_types and event.type not in self.event_types:
            return False
        
        if self.sources and event.source not in self.sources:
            return False
        
        if self.correlation_ids and event.correlation_id not in self.correlation_ids:
            return False
        
        if self.custom_filter and not self.custom_filter(event):
            return False
        
        return True


class EventSubscription:
    """事件订阅"""
    
    def __init__(self, 
                 handler: EventHandler,
                 filter: Optional[EventFilter] = None,
                 priority: int = 0):
        self.handler = handler
        self.filter = filter or EventFilter()
        self.priority = priority
        self.created_at = datetime.now(timezone.utc)
        self.last_processed = None
        self.processed_count = 0
        self.error_count = 0


class EventStorage(ABC):
    """事件存储抽象接口"""
    
    @abstractmethod
    async def store_event(self, event: Event) -> None:
        """存储事件"""
        pass
    
    @abstractmethod
    async def get_events(self, 
                        event_types: Optional[List[EventType]] = None,
                        sources: Optional[List[str]] = None,
                        start_time: Optional[datetime] = None,
                        end_time: Optional[datetime] = None,
                        limit: int = 100) -> List[Event]:
        """获取事件"""
        pass
    
    @abstractmethod
    async def get_event_by_id(self, event_id: str) -> Optional[Event]:
        """根据ID获取事件"""
        pass


class MemoryEventStorage(EventStorage):
    """内存事件存储实现"""
    
    def __init__(self, max_events: int = 10000):
        self.max_events = max_events
        self.events: List[Event] = []
        self.events_by_id: Dict[str, Event] = {}
        self.lock = threading.RLock()
    
    async def store_event(self, event: Event) -> None:
        """存储事件"""
        with self.lock:
            self.events.append(event)
            self.events_by_id[event.id] = event
            
            # 保持最大事件数量限制
            if len(self.events) > self.max_events:
                removed_event = self.events.pop(0)
                if removed_event.id in self.events_by_id:
                    del self.events_by_id[removed_event.id]
    
    async def get_events(self, 
                        event_types: Optional[List[EventType]] = None,
                        sources: Optional[List[str]] = None,
                        start_time: Optional[datetime] = None,
                        end_time: Optional[datetime] = None,
                        limit: int = 100) -> List[Event]:
        """获取事件"""
        with self.lock:
            filtered_events = []
            
            for event in reversed(self.events):  # 最新的在前
                if event_types and event.type not in event_types:
                    continue
                
                if sources and event.source not in sources:
                    continue
                
                if start_time and event.timestamp < start_time:
                    continue
                
                if end_time and event.timestamp > end_time:
                    continue
                
                filtered_events.append(event)
                
                if len(filtered_events) >= limit:
                    break
            
            return filtered_events
    
    async def get_event_by_id(self, event_id: str) -> Optional[Event]:
        """根据ID获取事件"""
        with self.lock:
            return self.events_by_id.get(event_id)


class EventBus:
    """事件总线主类"""
    
    def __init__(self, storage: Optional[EventStorage] = None):
        self.storage = storage or MemoryEventStorage()
        self.subscriptions: List[EventSubscription] = []
        self.subscription_lock = threading.RLock()
        self.event_queue = asyncio.Queue()
        self.is_running = False
        self.worker_task: Optional[asyncio.Task] = None
        
        logger.info("事件总线已初始化")
    
    def subscribe(self, 
                  handler: EventHandler,
                  filter: Optional[EventFilter] = None,
                  priority: int = 0) -> None:
        """订阅事件"""
        subscription = EventSubscription(handler, filter, priority)
        
        with self.subscription_lock:
            self.subscriptions.append(subscription)
            # 按优先级排序，优先级高的在前
            self.subscriptions.sort(key=lambda s: s.priority, reverse=True)
        
        logger.info(f"已订阅事件处理器: {handler.__class__.__name__}")
    
    def unsubscribe(self, handler: EventHandler) -> None:
        """取消订阅"""
        with self.subscription_lock:
            self.subscriptions = [
                s for s in self.subscriptions 
                if s.handler != handler
            ]
        
        logger.info(f"已取消订阅事件处理器: {handler.__class__.__name__}")
    
    async def publish(self, 
                     event_type: EventType,
                     source: str,
                     data: Dict[str, Any],
                     correlation_id: Optional[str] = None,
                     metadata: Optional[Dict[str, Any]] = None) -> str:
        """发布事件"""
        event = Event(
            id=str(uuid.uuid4()),
            type=event_type,
            source=source,
            timestamp=datetime.now(timezone.utc),
            data=data,
            correlation_id=correlation_id,
            metadata=metadata
        )
        
        # 存储事件
        await self.storage.store_event(event)
        
        # 加入处理队列
        await self.event_queue.put(event)
        
        logger.debug(f"已发布事件: {event_type.value} from {source}")
        return event.id
    
    async def start(self) -> None:
        """启动事件总线"""
        if self.is_running:
            return
        
        self.is_running = True
        self.worker_task = asyncio.create_task(self._event_worker())
        logger.info("事件总线已启动")
    
    async def stop(self) -> None:
        """停止事件总线"""
        if not self.is_running:
            return
        
        self.is_running = False
        
        if self.worker_task:
            self.worker_task.cancel()
            try:
                await self.worker_task
            except asyncio.CancelledError:
                pass
        
        logger.info("事件总线已停止")
    
    async def _event_worker(self) -> None:
        """事件处理工作线程"""
        while self.is_running:
            try:
                # 等待事件，设置超时避免阻塞
                event = await asyncio.wait_for(
                    self.event_queue.get(), 
                    timeout=1.0
                )
                
                await self._process_event(event)
                
            except asyncio.TimeoutError:
                continue
            except Exception as e:
                logger.error(f"事件处理工作线程错误: {e}", exc_info=True)
    
    async def _process_event(self, event: Event) -> None:
        """处理单个事件"""
        with self.subscription_lock:
            subscriptions = self.subscriptions.copy()
        
        # 按优先级处理订阅者
        for subscription in subscriptions:
            try:
                # 检查过滤器
                if not subscription.filter.matches(event):
                    continue
                
                # 检查处理器是否能处理此事件类型
                if not subscription.handler.can_handle(event.type):
                    continue
                
                # 处理事件
                await subscription.handler.handle(event)
                
                # 更新统计信息
                subscription.last_processed = datetime.now(timezone.utc)
                subscription.processed_count += 1
                
            except Exception as e:
                subscription.error_count += 1
                logger.error(
                    f"事件处理器 {subscription.handler.__class__.__name__} "
                    f"处理事件 {event.type.value} 时出错: {e}",
                    exc_info=True
                )
    
    async def get_event_statistics(self) -> Dict[str, Any]:
        """获取事件统计信息"""
        with self.subscription_lock:
            subscriptions = self.subscriptions.copy()
        
        stats = {
            "total_subscriptions": len(subscriptions),
            "queue_size": self.event_queue.qsize(),
            "is_running": self.is_running,
            "subscriptions": []
        }
        
        for subscription in subscriptions:
            stats["subscriptions"].append({
                "handler": subscription.handler.__class__.__name__,
                "priority": subscription.priority,
                "created_at": subscription.created_at.isoformat(),
                "last_processed": subscription.last_processed.isoformat() if subscription.last_processed else None,
                "processed_count": subscription.processed_count,
                "error_count": subscription.error_count
            })
        
        return stats


# 全局事件总线实例
_global_event_bus: Optional[EventBus] = None


def get_event_bus() -> EventBus:
    """获取全局事件总线实例"""
    global _global_event_bus
    if _global_event_bus is None:
        _global_event_bus = EventBus()
    return _global_event_bus


def set_event_bus(event_bus: EventBus) -> None:
    """设置全局事件总线实例"""
    global _global_event_bus
    _global_event_bus = event_bus


# 便捷函数
async def publish_event(event_type: EventType,
                       source: str,
                       data: Dict[str, Any],
                       correlation_id: Optional[str] = None,
                       metadata: Optional[Dict[str, Any]] = None) -> str:
    """发布事件的便捷函数"""
    return await get_event_bus().publish(
        event_type, source, data, correlation_id, metadata
    )


def subscribe_to_events(handler: EventHandler,
                       filter: Optional[EventFilter] = None,
                       priority: int = 0) -> None:
    """订阅事件的便捷函数"""
    get_event_bus().subscribe(handler, filter, priority)
